package david.java.flink_sql.streamApiIntegration;


import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description:
 * @Author: ZhaoDawei
 * @Date: Create in 10:14 上午 2022/3/29
 */
public class S3_1_HandleInsertOnlyStream_2 {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);



        DataStream<User> dataStream = env.fromElements(
                new User("Alice", 4 ),
                new User("Bob", 6),
                new User("Alice", 10)
        );

        Table table = tableEnv.fromDataStream(dataStream);
        table.printSchema();
        // (
        //   `name` STRING,
        //   `score` INT
        // )

        Table table2 = tableEnv.fromDataStream(
                dataStream,
                Schema.newBuilder()
                        .column("f0", DataTypes.of(User.class))
                        .build())
                .as("user");
        table2.printSchema();


        Table table3 = tableEnv.fromDataStream(
                dataStream,
                Schema.newBuilder()
                        .column(
                                "f0",
                                DataTypes.STRUCTURED(
                                        User.class,
                                        DataTypes.FIELD("name", DataTypes.STRING()),
                                        DataTypes.FIELD("score", DataTypes.INT())))
                        .build())
                .as("User");
        table3.printSchema();

    }


    public static class User {
        public String name;

        public Integer score;


        // default constructor for DataStream API
        public User() {
        }

        // fully assigning constructor for Table API
        public User(String name, Integer score) {
            this.name = name;
            this.score = score;
        }
    }
}
